package org.example.wordCount;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author huangqihan
 * @date 2021/2/9
 */
public class WordCountByStream {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // We can set on every step.
        // env.setParallelism(3);

        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String host = parameterTool.get("host");
        int port = parameterTool.getInt("port");

        // Use NetCat to read the data source. nc -lp 7777
        DataStreamSource<String> stringDataSource = env.socketTextStream(host, port);
        DataStream<Tuple2<String, Integer>> dataStream = stringDataSource
                .flatMap(new WordCountByBatch.MyFlatMapper()).slotSharingGroup("red")
                // 按照当前数据的 hashCode 分组
                .keyBy(0)
                .sum(1)
                .setParallelism(2).slotSharingGroup("green");

        dataStream.print().setParallelism(1);
        env.execute();
    }

}
